#include <stdio.h>
#include "shmem.h"
#include "msg_queue_peer.h"
#include <string.h>
#include "cJSON.h"
#include "MQTTClient.h"
#include "time.h"
#include <pthread.h>
#include <unistd.h>
#include <sys/stat.h> // 添加目录创建支持

// 上报方式定义
#define REPORT_REFRESH 0 // 刷新上报
#define REPORT_TIMED 1   // 定时上报
#define REPORT_CHANGE 2  // 变化上报

// MQTT宏定义
#define ADDRESS "tcp://192.168.50.155:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC  "/app/control/down"
#define TOPIC1 "/app/data/up"
#define QOS 1
#define TIMEOUT 10000L

// 消息队列名称
#define MQ_NAME "topic"

//全局变量
volatile MQTTClient_deliveryToken deliveredtoken;
pthread_mutex_t shm_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mq_mutex = PTHREAD_MUTEX_INITIALIZER; // 消息队列互斥锁
int report_interval = 5;          // 默认上报间隔5秒
int initial_change_reported = 0;    // 标记变化上报是否已完成初始上报
int current_report_mode = REPORT_CHANGE; // 当前上报模式，默认为变化上报
int refresh_requested = 0;          // 刷新上报请求标志
int rc;
int ch;

union val_t
{
    int b_val;   // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
};

// 设备类型定义
#define DEV_TYPE_STM32 0
#define DEV_TYPE_MODBUS 1
//消息队列参数
struct msgbuf
{
    long mtype;
    int value;
    int key;
    //char dev_type[256];
};
// 共享内存参数
static struct shm_param para;
struct std_node
{
    int key;             // 唯一键值
    int type;            // 数据点类型
    int dev_type;        // 数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; // 变化上报后需要更新旧值
    union val_t new_val; // 从共享内存取出最新数据，放到new_val中
    int ret;             // 默认为-1，采集成功后设置为0，采集失败再置-1
};
static int *total;                // 指向共享内存中数据节点总个数
static struct std_node *node_arr; // 指向共享内存中节点缓存数组头
static MQTTClient client;         //创建MQTT客户端

#define MAX_NODE 128 // 最大支持数目，实际可用是127
#define STD_NODE_LEN sizeof(struct std_node)
#define MAX_NODE_SIZE (MAX_NODE * STD_NODE_LEN)

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    // 尝试解析JSON消息
    cJSON *root = cJSON_Parse(message->payload);
    if (root)
    {
        printf("   JSON content:\n");
        char *json_str = cJSON_Print(root);
        if (json_str) {
            printf("%s\n", json_str);
            free(json_str);
        }

        // 1. 检查上报模式设置命令
        cJSON *mode = cJSON_GetObjectItem(root, "set_report_mode");
        if (mode && mode->type == cJSON_String) 
        {
            printf("Received report mode setting command\n");
            
            if (strcmp(mode->valuestring, "refresh") == 0) {
                current_report_mode = REPORT_REFRESH;
                printf("Report mode set to: REFRESH\n");
            } 
            else if (strcmp(mode->valuestring, "timed") == 0) {
                current_report_mode = REPORT_TIMED;
                printf("Report mode set to: TIMED\n");
                
                // 检查是否有上报间隔设置
                cJSON *interval = cJSON_GetObjectItem(root, "interval");
                if (interval && interval->type == cJSON_Number) {
                    report_interval = interval->valueint;
                    printf("Report interval set to: %d seconds\n", report_interval);
                }
            }
            else if (strcmp(mode->valuestring, "change") == 0) {
                current_report_mode = REPORT_CHANGE;
                initial_change_reported = 0;
                printf("Report mode set to: CHANGE\n");
            }
            else {
                printf("Unknown report mode: %s\n", mode->valuestring);
            }
        }

        // 2. 检查数据更新请求命令
        cJSON *report_type = cJSON_GetObjectItem(root, "report_type");
        if (report_type && report_type->type == cJSON_String && 
            strcmp(report_type->valuestring, "data_updata") == 0) 
        {
            printf("Received data update request\n");
            
            if (current_report_mode == REPORT_REFRESH) {
                refresh_requested = 1;
                printf("Refresh requested flag set (refresh mode)\n");
            }
            else {
                printf("Data update request ignored (current mode: %d)\n", current_report_mode);
            }
        }

        // 3. 检查数据点更新消息
        cJSON *data = cJSON_GetObjectItem(root, "data");
        if (data && data->type == cJSON_Array) 
        {
            printf("Received data point update message\n");
            int data_count = cJSON_GetArraySize(data);
            printf("Processing %d data points\n", data_count);
            
            for (int i = 0; i < data_count; i++) 
            {
                cJSON *item = cJSON_GetArrayItem(data, i);
                if (!item) {
                    printf("Error: Array item %d is null\n", i);
                    continue;
                }
                
                // 解析数据点
                cJSON *key_item = cJSON_GetObjectItem(item, "key");
                cJSON *value_item = cJSON_GetObjectItem(item, "value");
                
                // 验证必要字段
                if (key_item && key_item->type == cJSON_Number) 
                {
                    struct msgbuf msg_data;
                    msg_data.mtype = 1;
                    msg_data.key = key_item->valueint;
                    
                    // 处理数值类型
                    if (value_item && value_item->type == cJSON_Number) {
                        // 直接获取整数值
                        msg_data.value = value_item->valueint;
                    } else {
                        printf("Missing or invalid value for key %d\n", key_item->valueint);
                        continue;
                    }
                    
                    // 发送到消息队列
                    pthread_mutex_lock(&mq_mutex);
                    int send_ret = msg_queue_send(MQ_NAME, &msg_data, sizeof(msg_data), 0);
                    pthread_mutex_unlock(&mq_mutex);
                    
                    if (send_ret == 0) {
                        printf("Sent to message queue: key=%d, value=%d\n", 
                               msg_data.key, msg_data.value);
                    } else {
                        perror("Message queue send failed");
                    }
                }
                else 
                {
                    printf("Invalid data point format\n");
                    char *item_str = cJSON_Print(item);
                    if (item_str) {
                        printf("Invalid item: %s\n", item_str);
                        free(item_str);
                    }
                }
            }
        }

        // 删除JSON对象
        cJSON_Delete(root);
    }
    else
    {
        printf("   Raw message: %.*s\n", message->payloadlen, (char *)message->payload);
        printf("   Ignored non-JSON message\n");
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// 数据上报函数
// 修改后的数据上报函数（支持浮点数）
void send_report(struct std_node *nodes, int count, int report_type)
{
    cJSON *root = cJSON_CreateObject();
    if (!root) {
        fprintf(stderr, "Failed to create JSON root\n");
        return;
    }

    // 添加报告类型
    const char *type_str = "";
    switch (report_type) {
        case REPORT_REFRESH: type_str = "refresh"; break;
        case REPORT_TIMED:   type_str = "timed";   break;
        case REPORT_CHANGE: type_str = "change";  break;
        default:            type_str = "unknown";
    }
   // cJSON_AddStringToObject(root, "report_type", type_str);

    // 添加数据点数组
    cJSON *data_array = cJSON_CreateArray();
    if (!data_array) {
        cJSON_Delete(root);
        return;
    }

    for (int i = 0; i < count; i++) {
        // 只处理采集成功的数据点
        if (nodes[i].ret == 0) {
            cJSON *item = cJSON_CreateObject();
            if (!item) continue;
            
            // 添加key
            cJSON_AddNumberToObject(item, "key", nodes[i].key);
            
            // 根据数据类型添加value
            if (nodes[i].type == 0) { // bool类型
                cJSON_AddNumberToObject(item, "value", nodes[i].new_val.b_val);
            } else if (nodes[i].type == 1) { // int类型
                cJSON_AddNumberToObject(item, "value", nodes[i].new_val.i_val);
            } else if (nodes[i].type == 2) { // float类型
                cJSON_AddNumberToObject(item, "value", nodes[i].new_val.f_val);
            }
            
            cJSON_AddItemToArray(data_array, item);
        }
    }
    
    cJSON_AddItemToObject(root, "data", data_array);

    // 发布配置
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    // 序列化并发送
    char *json_str = cJSON_Print(root);
    if (!json_str) {
        cJSON_Delete(root);
        return;
    }
    
    printf("Generated JSON: %s\n", json_str);
    
    // 发布JSON消息
    pubmsg.payload = json_str;
    pubmsg.payloadlen = (int)strlen(json_str);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;

    MQTTClient_publishMessage(client, TOPIC1, &pubmsg, &token);
    printf("Publishing to topic: %s\n", TOPIC1);
    int rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered (rc=%d)\n", token, rc);

    // 清理资源
    free(json_str);
    cJSON_Delete(root);
}

// 上报模式处理线程
void *report_handler_thread(void *arg)
{
    printf("Report handler thread started\n");
    sleep(1);
    while (1) {
        pthread_mutex_lock(&shm_mutex);
        
        switch (current_report_mode) {
            case REPORT_REFRESH:
                // 刷新上报模式
                if (refresh_requested) {
                    send_report(node_arr, *total, REPORT_REFRESH);
                    refresh_requested = 0;
                }
                break;
                
            case REPORT_TIMED:{
                // 定时上报模式
                static time_t last_report_time = 0; //定时标志
                time_t now = time(NULL);
                
                if (difftime(now, last_report_time) >= report_interval) {
                    send_report(node_arr, *total, REPORT_TIMED);
                    last_report_time = now;
                }
                break;
            }
            case REPORT_CHANGE:
                // 变化上报模式
                if (!initial_change_reported) {
                    // 首次变化上报，发送全量数据
                    send_report(node_arr, *total, REPORT_CHANGE);
                    
                    // 初始化旧值
                    for (int i = 0; i < *total; i++) {
                        if (node_arr[i].ret == 0) {
                            memcpy(&node_arr[i].old_val, &node_arr[i].new_val, sizeof(union val_t));
                        }
                    }
                    initial_change_reported = 1;
                } else {
                    // 变化检测
                    struct std_node changed_nodes[MAX_NODE];
                    int changed_count = 0;
                    
                    for (int i = 0; i < *total; i++) {
                        // 不再跳过ret != 0的节点
                        if (memcmp(&node_arr[i].old_val, &node_arr[i].new_val, sizeof(union val_t)) != 0) {
                            changed_nodes[changed_count] = node_arr[i];
                            changed_count++;
                            
                            // 更新旧值
                            memcpy(&node_arr[i].old_val, &node_arr[i].new_val, sizeof(union val_t));
                        }
                    }
                    
                    // 上报变化点
                    if (changed_count > 0) {
                        send_report(changed_nodes, changed_count, REPORT_CHANGE);
                    }
                }
                break;
        }
        
        pthread_mutex_unlock(&shm_mutex);
        usleep(500000); // 500ms轮询
    }
    return NULL;
}

// 接收数据线程
void *recv_handler_thread(void *arg)
{
    printf("Recv handler thread started\n");
    while (1) {
        MQTTClient_message *message = NULL;
        char *topic = NULL;
        int topicLen;
        
        rc = MQTTClient_receive(client, &topic, &topicLen, &message, 1000);
        if (rc == MQTTCLIENT_SUCCESS && message) {
            // 处理接收到的消息
            printf("Received message on topic: %.*s\n", topicLen, topic);
            msgarrvd(NULL, topic, topicLen, message);
            
            // 释放资源
            MQTTClient_free(topic);
            MQTTClient_freeMessage(&message);
        }
    }
    return NULL;
}

// 解析JSON并映射到共享内存
// 修改后的parse_and_map_json函数
int parse_and_map_json(const char *json_str)
{
    pthread_mutex_lock(&shm_mutex);
    cJSON *root = cJSON_Parse(json_str);
    if (!root)
    {
        fprintf(stderr, "JSON parse error: %s\n", cJSON_GetErrorPtr());
        pthread_mutex_unlock(&shm_mutex);
        return -1;
    }

    // 获取设备节点
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    
    if (!stm32 || !modbus) {
        fprintf(stderr, "Missing device sections in JSON\n");
        cJSON_Delete(root);
        pthread_mutex_unlock(&shm_mutex);
        return -1;
    }

    // 获取数据数组 - 不使用 cJSON_IsArray
    cJSON *stm32_data = cJSON_GetObjectItem(stm32, "data");
    cJSON *modbus_data = cJSON_GetObjectItem(modbus, "data");
    
    // 检查数据节点是否存在
    if (!stm32_data || !modbus_data) {
        fprintf(stderr, "Missing data arrays in JSON\n");
        cJSON_Delete(root);
        pthread_mutex_unlock(&shm_mutex);
        return -1;
    }

    // 获取数组大小
    int stm32_count = cJSON_GetArraySize(stm32_data);
    int modbus_count = cJSON_GetArraySize(modbus_data);
    int total_nodes = stm32_count + modbus_count;

    if (total_nodes > MAX_NODE)
    {
        fprintf(stderr, "Error: Too many nodes (%d > %d)\n", total_nodes, MAX_NODE);
        cJSON_Delete(root);
        pthread_mutex_unlock(&shm_mutex);
        return -1;
    }

    // 设置节点总数
    *total = total_nodes;

    int index = 0;

    // 处理STM32节点
    for (int i = 0; i < stm32_count; i++)
    {
        cJSON *item = cJSON_GetArrayItem(stm32_data, i);
        if (!item) continue;
        
        cJSON *key_item = cJSON_GetObjectItem(item, "key");
        cJSON *type_item = cJSON_GetObjectItem(item, "type");
        
        if (!key_item || !type_item) {
            fprintf(stderr, "Missing key or type in STM32 node\n");
            continue;
        }

        struct std_node *node = &node_arr[index++];
        node->key = key_item->valueint;
        
        // 类型转换：JSON类型 -> 内部类型
        switch (type_item->valueint) {
            case 1: node->type = 0; break; // bool -> 0
            case 2: node->type = 1; break; // int -> 1
            case 3: node->type = 2; break; // float -> 2
            default: node->type = 1; // 默认为int
        }
        
        node->dev_type = DEV_TYPE_STM32;
        node->ret = 0;
        memset(&node->old_val, 0, sizeof(union val_t));
        memset(&node->new_val, 0, sizeof(union val_t));
    }

    // 处理Modbus节点
    for (int i = 0; i < modbus_count; i++)
    {
        cJSON *item = cJSON_GetArrayItem(modbus_data, i);
        if (!item) continue;
        
        cJSON *key_item = cJSON_GetObjectItem(item, "key");
        cJSON *type_item = cJSON_GetObjectItem(item, "type");
        
        if (!key_item || !type_item) {
            fprintf(stderr, "Missing key or type in Modbus node\n");
            continue;
        }

        struct std_node *node = &node_arr[index++];
        node->key = key_item->valueint;
        
        // 类型转换：JSON类型 -> 内部类型
        switch (type_item->valueint) {
            case 1: node->type = 0; break; // bool -> 0
            case 2: node->type = 1; break; // int -> 1
            case 3: node->type = 2; break; // float -> 2
            default: node->type = 1; // 默认为int
        }
        
        node->dev_type = DEV_TYPE_MODBUS;
        node->ret = 0;
        memset(&node->old_val, 0, sizeof(union val_t));
        memset(&node->new_val, 0, sizeof(union val_t));
    }

    cJSON_Delete(root);
    pthread_mutex_unlock(&shm_mutex);
    
    printf("Parsed %d STM32 nodes and %d Modbus nodes\n", stm32_count, modbus_count);
    return 0;
}

int main(int argc, char const *argv[])
{
    // 创建写共享内存**************************
    int ret = -1;
    ret = shm_init(&para, "shm_test", 4096);
    if (ret < 0)
    {
        return -1;
    }
    void *node_p = shm_getaddr(&para);
    if (node_p == NULL)
    {
        return -1;
    }
    // 设置共享内存指针
    total = (int *)node_p;
    node_arr = (struct std_node*)(node_p + sizeof(int));
    *total = 0; // 初始化为0个节点
    // 解析点表**************************
    // 打开文件
    FILE *file = fopen("node.json", "r");
    if (file == NULL)
    {
        perror("open err\n");
        return -1;
    }

    char buf[1024] = {0};

    size_t len = fread(buf, 1, 1024, file);
    if (len < 0)
    {
        perror("read err\n");
        return -1;
    }
   
    // 打印文件内容
    printf("%s\n", buf);
    fclose(file);
    // 解析JSON并映射到共享内存
    if (parse_and_map_json(buf))
    {
        fprintf(stderr, "Failed to parse and map JSON\n");
        return -1;
    }

    // 打印调试信息
    printf("Successfully mapped %d nodes to shared memory\n", *total);
    printf("Node details:\n");
    for (int i = 0; i < *total; i++)
    {
        struct std_node *node = &node_arr[i];
        printf("Node %d: key=%d, type=%d, dev_type=%s\n",
               i, node->key, node->type,
               node->dev_type == DEV_TYPE_STM32 ? "STM32" : "Modbus");
    }

    /// MQTT初始化******************************
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    
    // 在连接前设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    
    MQTTClient_subscribe(client, TOPIC, QOS);
    
    // 创建多线程*****************************
    pthread_t tid[2];
    if (pthread_create(&tid[0], NULL, report_handler_thread, NULL)) 
    {
        perror("Failed to create report thread");
    }
    if (pthread_create(&tid[1], NULL, recv_handler_thread, NULL)) 
    {
        perror("Failed to create receive thread");
    }
    
    // 主线程等待
    while(1) {
        printf("Press 'q' to quit: ");
        int ch = getchar();
        getchar(); // 消耗换行符
        
        if (ch == 'q') break;
    }
    
    // 清理资源
    MQTTClient_disconnect(client, 1000);
    MQTTClient_destroy(&client);
    shm_del(&para);
    return 0;
}